#! /usr/bin/env python3
# -*- coding: utf-8 -*-

from pykafka import KafkaClient
from pykafka.common import OffsetType


def get_kafka_data(hosts, topic, consumer_group):
    client = KafkaClient(hosts=hosts)
    topic = client.topics[topic]
    # consumer = topic.get_simple_consumer()
    # for message in consumer:
    #     if message is not None:
    #         print(message.offset, message.value)
    consumer = topic.get_balanced_consumer(
        consumer_group = consumer_group,
        # auto_commit_enable = True,
        auto_offset_reset = OffsetType.LATEST,
        reset_offset_on_start = True,
        use_rdkafka = False,
        managed = True
    )
    consumer.consume()
    consumer.commit_offsets()
    for message in consumer:
        if message is not None:
            # print('msg:{},offset:{}'.format(message.value,consumer.held_offsets))
            yield message.value


if __name__ == "__main__":
    hosts = "10.10.205.90:9092"
    topic = "dummyTopic"
    consumer_group = "group_mytest"
    worker_num = 6
    data = get_kafka_data(hosts, topic, consumer_group)
    for msg in data:
        print(msg)
